SpringBoot项目中需要把日志记录到：ElasticSearch 并使用 ：Kibana 搜索、展示、查看日志。

修改SpringBoot中的配置

因为SpringBoot默认使用的日志框架是：：Logback，但是：Log4j2 性能可能更高就打算替换掉。

替换方法，修改项目maven的pom.xml配置文件：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
    <exclusions>
        <exclusion>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-logging</artifactId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-log4j2</artifactId>
</dependency>

因为需要把日志记录到Kafka，所以引入org.apache.kafka kafka-clients到项目中：

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.10.0.0</version>
</dependency>

因为日志在本地测试和线上的kafka的服务器需要分开，通常我们可以在resources目录下面创建：

• 本地的log4j2的配置文件名叫log4j2-dev.xml
• 正式线上的log4j2的配置文件名叫log4j2-prd.xml

然后我们可以修改SpringBoot的bootstrap.properties，ACTIVE_ENVIRONMENT为环境变量。

spring.profiles.active=${ACTIVE_ENVIRONMENT:dev}
logging.config=classpath:log4j2-${ACTIVE_ENVIRONMENT}.xml

log4j2-xxx.xml的内容例子如下：

<?xml version="1.0" encoding="UTF-8"?>
<Configuration status="warn">
    <Properties>
    </Properties>
    <Appenders>
        <Console name="STDOUT" target="SYSTEM_OUT">
            <PatternLayout pattern="%d %p %c{1.} %t %m%n"/>
        </Console>
         <Kafka name="KAFKA-LOGGER" topic="service-logs-test"><!--kafka topic-->
            <JsonLayout/> <!--JsonLayout：日志格式为json,方便在Elastic中处理-->
            <Property name="bootstrap.servers">10.2.200.26:9092</Property><!--kafka server的ip:port-->
            <Property name ="retries" >3</Property>
            <Property name ="linger.ms" >1000</Property>
            <Property name ="buffer.memory" > 10485760</Property>
        </Kafka>
        <Async name="ASYNC-KAFKA-LOGGER">
            <AppenderRef ref="KAFKA-LOGGER"/>
            <LinkedTransferQueue/>
        </Async>
    </Appenders>
    <Loggers>
        <logger name="demo.xx.com.test" level="info" additivity="false"><!--日志级别大于info都会被记录到Kafka-->
            <AppenderRef ref="KAFKA-LOGGER"/>
        </logger>
        <!-- Root表示所有Logger用Root中的Appender打印日志  -->
        <Root level="info">
            <AppenderRef ref="STDOUT"/>
            <AppenderRef ref="ASYNC-KAFKA-LOGGER"/>
        </Root>
    </Loggers>
</Configuration>

在LogStash中的配置

LogStash组成结构：
Logstash 通过管道进行运作，管道有两个必需的元素，输入和输出，还有一个可选的元素，过滤器。输入插件从数据源获取数据，过滤器插件根据用户指定的数据格式修改数据，输出插件则将数据写入到目的地。
因此，需要一个配置文件管理输入、过滤器和输出相关的配置。配置文件内容格式如下：

#　输入
input {
  ...
}

# 过滤器
filter {
  ...
}

# 输出
output {
  ...
}

所以我们创建一个文件到LogStash的配置文件：logstash.yml中path.config配置的扫描目录。

service-logs.conf配置文件内容如下：

input {
   kafka {
        topics_pattern => ["service-logs-.*"]
        bootstrap_servers => "10.1.220.27:9092"
        group_id => "local-logstash"
        decorate_events => true
   }
}

filter {
    json {
        source => "message"
    }
    mutate {
        add_field => {
                "kafka_topic" => "%{[@metadata][kafka][topic]}"
                "kafka_consumer_group" => "%{[@metadata][kafka][consumer_group]}"
                "kafka_partition" => "%{[@metadata][kafka][partition]}"
                "kafka_offset" => "%{[@metadata][kafka][offset]}"
                "kafka_timestamp" => "%{[@metadata][kafka][timestamp]}"
        }
    }
}

output {
  if [level] == "ERROR" {
        http {
                url => "http://localhost:8088/send/mail"
                http_method => "post"
                automatic_retries => 0
                connect_timeout => 6
                keepalive => false
        }
  }

  elasticsearch {
        hosts => ["127.0.0.1:9200"]
        action => "index"
        index => "logstash-%{[kafka_topic]}"
  }
}

我们可以看到上面output针对日志级别是ERROR的会发送一个HTTP请求到一个发邮件的服务器，因为LogStash自带的邮件发不出来，所以单独用了nodejs写了一个发邮件接口。

如果使用LogStash自带的配置如下：

output {
      if [level] == "ERROR" {
            email {
                    debug => true
                    to => "xx@qq.com"
                    from => "xxx@126.com"
                    subject => 'Alert - Error'
                    codec => "plain"
                    contenttype => "text/html; charset=UTF-8"
                    body => "Tags: %{tags}\\n\\Content:\\n%{message}"
                    username => "xxx@126.com"
                    password => "xxx"
                    address => "smtp.126.com"
                    domain => "smtp.126.com"
                    port => 25
                    use_tls => false
                    via => "smtp"
            }
      }
  }

LogStash的功能很强大，更多可以阅读官方的文档。